package cn.cihon.ubi.qorosauto.storm;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class StormClusterTopology {

	public static void main(String[] args) throws Exception {
        
        //Topology definition
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("activemqspout",new ActiveMQSpout(),1);
		//builder.setBolt("rawdatastoragebolt", new RawDataStorageBolt(),1).shuffleGrouping("activemqspout");  //进行二进制数据的存储
		builder.setBolt("resolvebolt", new ResolveBolt(),1).shuffleGrouping("activemqspout");          //对二进制数据进行明文解析
		builder.setBolt("vehicleStandarddatastoragebolt", new VehicleStandardDataStorageBolt(),1).fieldsGrouping("resolvebolt", new Fields("VIN")); //对明文数据进行存储
		builder.setBolt("tripcalculatebolt", new TripCalculateBolt(),1).fieldsGrouping("resolvebolt", new Fields("VIN")); //对明文数据进行计算
		builder.setBolt("tripdatastoragebolt", new TripDataStorageBolt(),1).shuffleGrouping("tripcalculatebolt"); //对计算后数据进行存储
		builder.setBolt("redisstoragebolt", new RedisStorageBolt(),1).shuffleGrouping("tripcalculatebolt"); //对计算后数据进行Redis存储

        //Configuration
		Config conf = new Config();
		conf.setDebug(true);
		conf.setNumWorkers(1);
		conf.setMessageTimeoutSecs(300);
		
        // Topology run
		
		// conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
		// conf.put(Config.TOPOLOGY_DEBUG, true);
		StormSubmitter.submitTopology("Change-Topology", conf, builder.createTopology());
		// Submit topology for execution
//		try {
//			// Wait for some time before exiting
//			System.out.println("Waiting to consume from activeMQ");
//			Thread.sleep(10000);
//		} catch (Exception exception) {
//			System.out.println("Thread interrupted exception : " + exception);
//		}
		//cluster.killTopology("Change-Topology");
		//cluster.shutdown();
		
	}
}

